package com.py.custom.ability.kafka.producer;

import com.py.custom.ability.kafka.producer.base.ProducerProcessor;
import com.py.custom.ability.kafka.utils.MultiBinder;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.lang.NonNull;

import java.util.List;

/**
 * kafka生产者，多实例注册
 */
public class MultiProducerAutoDefinitionRegistrar implements ImportBeanDefinitionRegistrar,
        EnvironmentAware, BeanFactoryAware {

    private BeanFactory beanFactory;

    private Environment environment;


    @Override
    public void setBeanFactory(@NonNull BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

    @Override
    public void setEnvironment(@NonNull Environment environment) {
        this.environment = environment;
    }

    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        // 注册

        // 1.解析配置
        MultiBinder.PropertiesChecker<CKafkaProducerProperties> checker = buildProducerChecker();
        List<CKafkaProducerProperties> cKafkaProducerPropertiesList = new MultiBinder<>(environment, CKafkaProducerProperties.PREFIX, checker)
                .bind(CKafkaProducerProperties.class);

        // 2.注册 producer
        for (CKafkaProducerProperties cKafkaProducerProperties : cKafkaProducerPropertiesList) {
            // todo 封装成一个方法
            BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(CKafkaProducerProxy.class);
            beanDefinitionBuilder.addPropertyValue("cKafkaProducerProperties", cKafkaProducerProperties);
            beanDefinitionBuilder.setDestroyMethodName("close");
            beanDefinitionBuilder.applyCustomizers(
                    beanDefinition -> {
                        ((AbstractBeanDefinition) beanDefinition).setInstanceSupplier(() -> {
                            CKafkaProducerProxy<ProducerProcessor> ckafkaFactory = new CKafkaProducerProxy<>();
                            ckafkaFactory.setcKafkaProducerProperties(cKafkaProducerProperties);
                            return ckafkaFactory;
                        });
                    }
            );
            AbstractBeanDefinition beanDefinition = beanDefinitionBuilder.getBeanDefinition();

            String beanName = cKafkaProducerProperties.getProducerName();
            if (!registry.containsBeanDefinition(beanName) && !beanFactory.containsBean(beanName)) {
                registry.registerBeanDefinition(beanName, beanDefinition);
            }
        }
    }


    private MultiBinder.PropertiesChecker<CKafkaProducerProperties> buildProducerChecker() {
        return new MultiBinder.PropertiesChecker<CKafkaProducerProperties>() {
            @Override
            public boolean check(CKafkaProducerProperties toBeCheckedProperties) {
                if (StringUtils.isBlank(toBeCheckedProperties.getBrokerServers())) {
                    return false;
                }

                if (StringUtils.isBlank(toBeCheckedProperties.getTopicName())) {
                    return false;
                }
                // todo 事物消息
                // toBeCheckedProperties.getHasTransaction();
                return true;
            }
        };
    }

}
